package com.youxin.dataStream.checkPoint;


import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.IOException;

/**
 * 状态的后端存储
 * 默认情况下，state会保存在taskManager的内存中，checkpoint会存储在jobManager的内存中。
 *
 * state的store和checkpoint的位置取决于state backend的位置
 *      env.setStateBackend()
 *
 * 一共有三种state Backend
 *      MenoryStateBackend
 *          state数据保存在java堆内存中，执行checkpoint的时候，会把state的快照数据保存到jobManager的内存中
 *      FsStaeBackend
 *          state数据保存在taskManager的内存中，当执行checkpoint的时候会把state的快照数据保存到配置的文件系统中
 *          可以使用hdfs等分布式文件系统
 *      RocksDBStateBackend（适合在生产中使用）
 *          他会在本地的文件系统中委会状态，state会直接写入rocksdb中，同时他需要配置一个远端的文件系统（可以hdfs），
 *          在做checkpoint的时候会把本地的数据直接复制到远端的文件系统中。fail over的时候从filesystem中恢复到本地
 *
 *
 *
 *
 */
public class StateBackendText {
    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(1000);
        //单个任务修改checkpoint的方式
        //RocksDB 需要添加依赖
        env.setStateBackend(new RocksDBStateBackend("hdfs://hadoop-1:8020/flink/checkpoint",true));//递增的checkpoint
        env.setStateBackend(new FsStateBackend("hdfs://hadoop-1:8020/flink/checkpoints"));


        /**
         * 全局调整
         * 修改 flink-conf.yaml
         * state.backend: rocksdb
         * state.checkpoints.dir: hdfs://hadoop-1:8020/flink/checkpoints
         *
         * state.backend的值可以是下面几种：jobmanager(MemoryStateBackend), filesystem(FsStateBackend), rocksdb(RocksDBStateBackend)
         */




    }
}
